package felix.things.router.mqtt.persistent;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.type.CollectionType;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.io.*;
import java.util.*;
import java.util.concurrent.ConcurrentLinkedDeque;

/**
 * @author Felix
 * @date 2019/7/27
 */
@Slf4j
@Service
public class PersistentFileServiceImpl implements PersistentFileService {

    private static final long DEFAULT_POLLING_INTERVAL = 1000;
    private static final int DEFAULT_BUFFER_SIZE = 2000;
    private static final String DEFAULT_PATH = "storage";

    private static final String STORAGE_FILE_PREFIX = "storage-";
    private static final String RESEND_FILE_PREFIX = "resend-";
    private static final String STORAGE_FILE_NAME_REGEX = STORAGE_FILE_PREFIX + "\\d+";
    private static final String RESEND_FILE_NAME_REGEX = RESEND_FILE_PREFIX + "\\d+";
    private static final String DASH = "-";

    @Autowired
    private PersistenceConfiguration configuration;

    private ConcurrentLinkedDeque<MqttPersistentMessage> sendBuffer;
    private ConcurrentLinkedDeque<MqttPersistentMessage> resendBuffer;

    private List<File> storageFiles;
    private List<File> resendFiles;

    private int storageFileCounter;
    private int resendFileCounter;

    private File storageDir;

    @PostConstruct
    public void init() {
        checkConfig();
        initStorageDir();
        initFiles();
        initFileCounters();
        initBuffers();
    }

    private void checkConfig() {
        if (configuration.getPollingInterval() == 0) {
            configuration.setPollingInterval(DEFAULT_POLLING_INTERVAL);
        }
        if (configuration.getBufferSize() == 0) {
            configuration.setBufferSize(DEFAULT_BUFFER_SIZE);
        }
        if (StringUtils.isEmpty(configuration.getPath())) {
            configuration.setPath(DEFAULT_PATH);
        }
        if (configuration.getStorageSize() == 0) {
            configuration.setStorageSize(Integer.MAX_VALUE);
        }
    }

    private void initStorageDir() {
        storageDir = new File(configuration.getPath());
        if (!storageDir.exists()) {
            storageDir.mkdirs();
        }
    }

    private void initFiles() {
        storageFiles = getFiles(STORAGE_FILE_NAME_REGEX);
        resendFiles = getFiles(RESEND_FILE_NAME_REGEX);
    }

    private List<File> getFiles(String nameRegex) {
        File[] filesArray = storageDir.listFiles((file) -> !file.isDirectory() && file.getName().matches(nameRegex));
        Arrays.sort(filesArray, Comparator.comparing(File::lastModified));
        return new ArrayList<>(Arrays.asList(filesArray));
    }

    private void initFileCounters() {
        storageFileCounter = getFileCounter(storageFiles);
        resendFileCounter = getFileCounter(resendFiles);
    }

    private int getFileCounter(List<File> files) {
        int counter = 0;
        if (files.isEmpty()) {
            return counter;
        } else {
            String lastFileName = files.get(files.size() - 1).getName();
            int lastFileCounter = Integer.parseInt(lastFileName.substring(lastFileName.lastIndexOf(DASH) + 1));
            if (lastFileCounter >= configuration.getStorageSize()) {
                counter = 0;
            } else {
                counter = lastFileCounter + 1;
            }
            return counter;
        }
    }

    private void initBuffers() {
        sendBuffer = new ConcurrentLinkedDeque<>();
        resendBuffer = new ConcurrentLinkedDeque<>();
    }

    @Override
    public void persistMessage(String topic, byte[] payload) throws IOException {
        MqttPersistentMessage message = MqttPersistentMessage.builder().id(UUID.randomUUID())
                .timestamp(System.currentTimeMillis())
                .topic(topic).payload(payload).build();
        addMessageToBuffer(message);
    }

    @Override
    public List<MqttPersistentMessage> getPersistentMessages() throws IOException {
        return getMqttPersistentMessages(storageFiles, sendBuffer);
    }

    @Override
    public List<MqttPersistentMessage> getResendMessages() throws IOException {
        return getMqttPersistentMessages(resendFiles, resendBuffer);
    }

    private List<MqttPersistentMessage> getMqttPersistentMessages(List<File> files, ConcurrentLinkedDeque<MqttPersistentMessage> buffer) throws IOException {
        List<MqttPersistentMessage> messages;
        if (files.isEmpty()) {
            messages = new ArrayList<>(buffer);
            buffer.clear();
            return messages;
        }
        File oldestFile = files.remove(0);
        messages = readFromFile(oldestFile);
        oldestFile.delete();
        return messages;
    }

    /**
     * 重试发送
     *
     * @param message
     * @throws IOException
     */
    @Override
    public void saveForResend(MqttPersistentMessage message) throws IOException {
        if (resendBuffer.size() >= configuration.getBufferSize()) {
            if (resendFileCounter >= configuration.getStorageSize()) {
                resendFileCounter = 0;
            } else {
                resendFileCounter++;
            }
            File file = flushBufferToFile(resendBuffer, RESEND_FILE_PREFIX + resendFileCounter);
            boolean b = storageFiles.stream().anyMatch(u -> u.getName().equals(file.getName()));
            if (!b) {
                resendFiles.add(file);
            }
        }
        resendBuffer.add(message);
    }

    @Override
    public void saveForResend(List<MqttPersistentMessage> messages) throws IOException {
        for (MqttPersistentMessage message : messages) {
            saveForResend(message);
        }
    }

    /**
     * 根据bufferSize来判断是否存储文件
     *
     * @param message
     * @throws IOException
     */
    private void addMessageToBuffer(MqttPersistentMessage message) throws IOException {
        if (sendBuffer.size() >= configuration.getBufferSize()) {
            if (storageFileCounter >= configuration.getStorageSize()) {
                storageFileCounter = 0;
            } else {
                storageFileCounter++;
            }
            File file = flushBufferToFile(sendBuffer, STORAGE_FILE_PREFIX + storageFileCounter);
            boolean b = storageFiles.stream().anyMatch(u -> u.getName().equals(file.getName()));
            if (!b) {
                storageFiles.add(file);
            }
        }
        sendBuffer.add(message);
    }

    private File flushBufferToFile(ConcurrentLinkedDeque<MqttPersistentMessage> buffer, String fileName) throws IOException {
        ObjectOutputStream outStream = null;
        try {
            File newFile = new File(storageDir, fileName);
            outStream = new ObjectOutputStream(new FileOutputStream(newFile));
            for (MqttPersistentMessage message : buffer) {
                outStream.writeObject(message);
            }
            buffer.clear();
            return newFile;
        } catch (IOException e) {
            log.error(e.getMessage(), e);
            throw e;
        } finally {
            try {
                if (outStream != null) {
                    outStream.close();
                }
            } catch (IOException e) {
                log.error(e.getMessage(), e);
                throw e;
            }
        }
    }

    private List<MqttPersistentMessage> readFromFile(File file) throws IOException {
        List<MqttPersistentMessage> messages = new ArrayList<>();
        ObjectInputStream inputStream = null;
        try {
            inputStream = new ObjectInputStream(new FileInputStream(file));
            while (true) {
                MqttPersistentMessage p = (MqttPersistentMessage) inputStream.readObject();
                messages.add(p);
            }
        } catch (EOFException e) {
            return messages;
        } catch (ClassNotFoundException e) {
            log.error(e.getMessage(), e);
        } catch (IOException e) {
            log.error(e.getMessage(), e);
            throw e;
        } finally {
            try {
                if (inputStream != null) {
                    inputStream.close();
                }
            } catch (IOException e) {
                log.error(e.getMessage(), e);
                throw e;
            }
        }
        return messages;
    }

    @Override
    public void coverPersistentsToJson() throws IOException {
        ObjectMapper mapper = new ObjectMapper();
        coverPersistentsToJson(mapper, resendFiles);
        coverPersistentsToJson(mapper, storageFiles);
    }

    @Override
    public void coverJsonToPersistents() throws IOException {
        ObjectMapper mapper = new ObjectMapper();
        for (File resendFile : storageFiles) {
            String json = readJsonFile(resendFile.getName() + ".json");
            CollectionType collectionType = mapper.getTypeFactory().constructCollectionType(ArrayList.class,
                    MqttPersistentMessage.class);
            List<MqttPersistentMessage> mqttPersistentMessages = mapper.readValue(json, collectionType);
            System.out.println(mqttPersistentMessages);
        }
    }

    private void coverPersistentsToJson(ObjectMapper mapper, List<File> storageFiles) throws IOException {
        for (File storageFile : storageFiles) {
            List<MqttPersistentMessage> mqttPersistentMessages = readFromFile(storageFile);
            writeJsonFile(mapper.writerWithDefaultPrettyPrinter().writeValueAsString(mqttPersistentMessages),
                    storageFile.getName() + ".json");
        }
    }

    private String readJsonFile(String fileName) {
        StringBuilder json = new StringBuilder();
        BufferedReader reader;
        try {
            File newFile = new File(storageDir, fileName);
            reader = new BufferedReader(new FileReader(newFile));
            String tempString = null;
            while ((tempString = reader.readLine()) != null) {
                json.append(tempString);
            }
            reader.close();
            newFile.delete();
        } catch (IOException e1) {
            e1.printStackTrace();
        }
        return json.toString();
    }

    private void writeJsonFile(String jsonString, String fileName) {
        try {
            File newFile = new File(storageDir, fileName);
            FileWriter fw = new FileWriter(newFile);
            PrintWriter out = new PrintWriter(fw);
            out.write(jsonString);
            out.println();
            fw.close();
            out.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
